package com.storm.lesson.lesson;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Map;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

/**
 * Spout是拓扑的流的来源，是一个拓扑中产生元数据流的组件。通常情况下，Spout会从外部数据源中读取数据，然后转换为拓扑内部的源数据
 * 
 * Spout可以发出超过一个流。为此，使用OutputFieldsDeclare类的declareStream方法来声明多个流，
 * 使用SpoutOutputCollector类的emit执行流的提交
 * 
 * Spout的主要方法是nextTuple()。nextTuple()会发出一个新Tuple到拓扑，如果没有新的元组发出，则简单地返回。
 * 
 * @author zhihu
 *
 */
public class MySpout implements IRichSpout {

	private static final long serialVersionUID = 1L;

	FileInputStream fis;
	InputStreamReader isr;
	BufferedReader br;

	SpoutOutputCollector collector = null;

	String str = null;

	@Override
	public void ack(Object msgId) {
		System.out.println("spout ack: " + msgId.toString());
	}

	@Override
	public void activate() {
		// TODO Auto-generated method stub

	}

	@Override
	public void close() {

		try {
			br.close();
			isr.close();
			fis.close();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	@Override
	public void deactivate() {
		// TODO Auto-generated method stub

	}

	@Override
	public void fail(Object msgId) {
		System.out.println("spout fail: " + msgId.toString());
	}

	@Override
	public void nextTuple() {
		try {
			while ((str = this.br.readLine()) != null) {
				collector.emit(new Values(str));
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Override
	public void open(Map conf, TopologyContext context,
			SpoutOutputCollector collector) {
		try {
			this.collector = collector;
			this.fis = new FileInputStream("track.log");
			this.isr = new InputStreamReader(fis, "UTF-8");
			this.br = new BufferedReader(isr);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("log"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

}
